from collections import namedtuple, defaultdict
from datetime import datetime
import backtrader as bt
import pandas as pd
from apscheduler.events import EVENT_JOB_EXECUTED, EVENT_JOB_ERROR
from apscheduler.executors.pool import ThreadPoolExecutor, ProcessPoolExecutor
from apscheduler.jobstores.memory import MemoryJobStore
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.schedulers.blocking import BlockingScheduler
from backtrader import feed, feeds
import queue

from pyLibs import units
from pyLibs.Gadfly import StockStore, BacktraderUtits as btu

data_queue = dict()


class Scheduler:
    def __init__(self, config):
        self._state = None
        self.stocks = list()
        self.conf = config
        self.tdxStore = StockStore.TdxStore(config)
        job = {
            'default': {'coalesce': False, 'max_instances': 3},
            'store': {'default': MemoryJobStore()},
        }
        if config.Scheduler.pool.capitalize() == 'Thread':  # 线程池执行器
            self.scheduler = ThreadPoolExecutor()
            job['executor'] = {'default': ThreadPoolExecutor(config.Scheduler.pool_num)}
        elif config.Scheduler.pool.capitalize() == 'Process':  # 进程池执行器
            self.scheduler = ProcessPoolExecutor()
            job['executor'] = {'default': ProcessPoolExecutor(config.Scheduler.pool_num)}
        if config.Scheduler.executor_type.capitalize() == 'Background':
            self.scheduler = BackgroundScheduler(executors=job['executor'],
                                                 job_defaults=job['default'],
                                                 jobstores=job['store'])
        else:
            self.scheduler = BlockingScheduler(executors=job['executor'],
                                               job_defaults=job['default'],
                                               jobstores=job['store'])


    '''
    stock
        dict: {'stock_code': '证券代码', 'cycle': '分析周期', 'bars': '初始K线数量', 'fromdate': '开始', 'todate': '结束'}
        tuple: ('stock_code', 'cycle', 'bars', 'fromdate', 'todate')
    '''
    def start(self, cerebro, codes):
        for code in codes:
            info = StockStore.get_stock_market(code)
            data_queue[info.tdx_code] = queue.Queue()
            cycle = StockStore.get_stock_cycle(self.conf.Scheduler.store.cycle)
            Stock = namedtuple('Stock', ['stock_code', 'cycle', 'info'])
            stock = Stock(code, cycle, info)
            self.stocks.append(stock)
            history = self.tdxStore.get_stock_data(code, cycle.cycle, cycle.bars).data
            cerebro.adddata(btu.StockLiveData(stock, dataname=history, name=info.tdx_code, timeframe=cycle.timeframe,
                                              compression=cycle.compression,
                                              fromdate=self.conf.Scheduler.store.fromdate,
                                              todate=self.conf.Scheduler.store.todate))
        self.scheduler.add_listener(self.listener_execute, EVENT_JOB_EXECUTED | EVENT_JOB_ERROR)
        self.scheduler.add_job(self.job_executes, 'cron', hour='09-23', minute='00-59', second='0')
        self.scheduler.start()
        cerebro.run()

    def job_executes(self):
        for ci in self.stocks:
            data_queue[ci.info.tdx_code].put(self.tdxStore.getStockBarsFromApi(ci.cycle.category, ci.stock_code, 0, 2, True))

    # 定义函数监听事件
    def listener_execute(self, event):
        if event.exception:
            print('The job did not run')
        else:
            print('The job is run！')


'''class StockDirectFeed(bt.feeds.PandasData):
    # States for the Finite State Machine in _load
    _ST_START, _ST_LIVE, _ST_HISTORBACK, _ST_OVER = range(4)
    lines = ('ratio')  # 要添加的线
    params = (
        ('nullvalue', 0.0),
        ('ratio', -1),
    )

    def __init__(self, stock):
        super(StockDirectFeed, self).__init__()
        self._state = self._ST_LIVE
        self.stock = stock

    def start(self):
        super(StockDirectFeed, self).start()
        for row in self.params.dataname.itertuples(index=True, name='Stock'):
            rows = row._asdict()
            rows.pop('Index')
            bf = pd.DataFrame(rows, index=[0])
            bf.index = pd.to_datetime(bf.datetime)
            data_queue[self.stock.info.tdx_code].put(bf)
            #self._set_lines(bf)
        #self._state = self._ST_HISTORBACK

    def _load(self):
        if self._state == self._ST_OVER:
            return False
        while True:
            try:
                if not data_queue[self.stock.info.tdx_code]:
                    return None
                data = data_queue[self.stock.info.tdx_code].get()
                if 'datetime' in data.keys():
                    dtime = datetime.strptime(data['datetime'].values[0], '%Y-%m-%d %H:%M')
                else:
                    dtime = datetime.now()
                self.lines.datetime[0] = bt.date2num(dtime)
                if 'open' in data.keys():
                    self.lines.open[0] = data['open'].values[0]
                if 'high' in data.keys():
                    self.lines.high[0] = data['high'].values[0]
                if 'low' in data.keys():
                    self.lines.low[0] = data['low'].values[0]
                if 'close' in data.keys():
                    self.lines.close[0] = data['close'].values[0]
                if 'ratio' in data.keys():
                    self.lines.ratio[0] = data['ratio'].values[0]
                if 'amount' in data.keys():
                    self.lines.amount[0] = data['amount'].values[0]
                if 'volume' in data.keys():
                    self.lines.volume[0] = data['volume'].values[0]
                if 'openinterest' in data.keys():
                    self.lines.openinterest[0] = data['openinterest'].values[0]
            except (KeyError, TypeError) as e:
                print('Bad Syntax in alert. Please check')
                print('{}'.format(e))
                print('Data Supplied: {}'.format(data))
                return False
            return True
        return None


    def _set_lines(self, data):
        if 'datetime' in data.keys():
            dtime = datetime.strptime(data['datetime'].values[0], '%Y-%m-%d %H:%M')
        else:
            dtime = datetime.now()
        self.lines.datetime[0] = bt.date2num(dtime)
        if 'open' in data.keys():
            self.lines.open[0] = data['open'].values[0]
        if 'high' in data.keys():
            self.lines.high[0] = data['high'].values[0]
        if 'low' in data.keys():
            self.lines.low[0] = data['low'].values[0]
        if 'close' in data.keys():
            self.lines.close[0] = data['close'].values[0]
        if 'ratio' in data.keys():
            self.lines.ratio[0] = data['ratio'].values[0]
        if 'amount' in data.keys():
            self.lines.amount[0] = data['amount'].values[0]
        if 'volume' in data.keys():
            self.lines.volume[0] = data['volume'].values[0]
        if 'openinterest' in data.keys():
            self.lines.openinterest[0] = data['openinterest'].values[0]

    def haslivedata(self):
        return self._state == self._ST_LIVE

    def islive(self):
        return True



class DataFeed(feed.DataBase):
    # States for the Finite State Machine in _load
    _ST_START, _ST_LIVE, _ST_HISTORBACK, _ST_OVER = range(4)

    def __init__(self, code, timeframe, compression=1, bars=250, fromdate=None, todate=None):
        self._state = None
        info = StockStore.get_stock_market(code)
        self.p.name = code
        self.p.dataname = info.tdx_code
        self.p.timeframe = timeframe
        self.p.compression = compression
        self.p.fromdate = fromdate
        self.p.todate = todate
        self.tdxStore = StockStore.TdxStore()
        self.data_queue = dict()
        self.currentData = dict()
        self.bars = bars
        CodeInfo = namedtuple('CodeInfo', ['stock_code', 'market', 'securities', 'tdx_code', 'cycle'])
        self.code = CodeInfo(code, info.market, info.securities, info.tdx_code, StockStore.get_stock_cycle(timeframe))

    def start(self):
        feed.DataBase.start(self)
        for ci in self.codeInfo:
            historys = self.tdxStore.get_stock_data(ci.stock_code, ci.cycle.cycle, self.bars).data
            self.data_queue[ci.tdx_code] = queue.Queue()
            self.p.dataname = ci.tdx_code
            for i in range(0, len(historys)):
                history = historys.iloc[i].to_dict()
                dt = datetime.strptime(history['datetime'], "%Y-%m-%d %H:%M")
                history['datetime'] = dt
                history['open'] = float(history['open'])
                history['high'] = float(history['high'])
                history['low'] = float(history['low'])
                history['close'] = float(history['close'])
                history['volume'] = int(history['volume'])
                history['amount'] = float(history['amount'])
                history['ratio'] = float(history['ratio'])
                history["openinterest"] = 0
                self.data_queue[ci.tdx_code].put(history)
            print('回填【' + ci.tdx_code + '】历史数据：', len(historys) + ' Bars')
        # 给EVENT_JOB_EXECUTED[执行完成job事件]添加回调，这里就是每次Job执行完成了我们就输出一些信息
        self.scheduler.add_listener(self.listener_execute, EVENT_JOB_EXECUTED | EVENT_JOB_ERROR)
        self.scheduler.add_job(self.job_executes, 'cron', day_of_week='mon-fri', hour='09-23',
                               minute='00-59', second='0')
        self.scheduler.start()
        self._state = self._ST_LIVE

    def job_executes(self):
        for code in self.codeInfo:
            category = code.cycle.category
            stock_code = code.stock_code
            self.data_queue[code.tdx_code] = self.tdxStore.getStockBarsFromApi(category, stock_code, 0, 2)

    # ***********               关于 APScheduler中有关Event相关使用示例               *************
    # 定义函数监听事件
    def listener_execute(self, event):
        if event.exception:
            print('The job did not run')
        else:
            print('The job is run！')

    def _load(self):
        if self._state == self._ST_OVER:
            return False
        while True:
            for p in self.codeInfo:
                if not self.data_queue[p.tdx_code]:
                    return None
                data = self.data_queue[p.tdx_code].get()
                self.params.name = p.tdx_code
                try:
                    self.lines.datetime[0] = bt.date2num(data['datetime'])
                    if 'open' in data.keys():
                        self.lines.open[0] = data['open']
                    if 'high' in data.keys():
                        self.lines.high[0] = data['high']
                    if 'low' in data.keys():
                        self.lines.low[0] = data['low']
                    if 'close' in data.keys():
                        self.lines.close[0] = data['close']
                    if 'volume' in data.keys():
                        self.lines.volume[0] = data['volume']
                    if 'openinterest' in data.keys():
                        self.lines.openinterest[0] = data['openinterest']
                except (KeyError, TypeError) as e:
                    print('Bad Syntax in alert. Please check')
                    print('{}'.format(e))
                    print('Data Supplied: {}'.format(data))
                    return False
                return True
        return None

    def haslivedata(self):
        return self._state == self._ST_LIVE

    def islive(self):
        return True'''
